package com.sunday.common.mq.rabbit.study.base.e12_DeadLetterExchanges;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * 死信交换
 * 来自队列的消息可以是“死信”;也就是说，重新发布到 发生以下任何事件时的交换：
 * <p>
 * 1.使用者使用 basic.reject 或 basic.nack 对消息进行否定确认，并将重新排队参数设置为 false。
 * 2.消息由于每条消息的 TTL 而过期;或
 * 3.邮件被丢弃，因为其队列超出了长度限制
 * <p>
 * 请注意，队列过期不会对其中的消息进行死信。
 * 死信交换 （DLX） 是正常的交换。他们可以是 任何常用类型，并按常规声明。
 * 对于任何给定的队列，DLX 可以由客户端使用队列的参数定义，也可以在服务器中定义 使用策略。在 策略和参数都指定 DLX 的情况，即 在参数中指定会否决策略中指定的参数。
 * 建议使用策略进行配置，因为它允许 DLX 不涉及应用程序重新部署的重新配置。
 * <p>
 * 路由死信消息
 * 死信消息被路由到其死信 交换以下任一：
 * 使用为队列指定的路由密钥，它们 开;或者，如果未设置，
 * 使用与最初相同的路由密钥 发布者
 * 例如，如果您将消息发布到具有 路由键 foo，该消息是 死信，它将被发布到它的死信 使用路由键 foo 进行交换。如果队列 最初登陆的消息已声明为 x-dead-letter-routeing-key 设置为 bar，然后消息将发布到 它的死信与路由键栏交换。
 * 请注意，如果未为 队列，其上的消息是死信的，具有所有原始路由密钥。这包括路由密钥 由 CC 和密件抄送标头添加 （有关这两个标头的详细信息，请参阅发件人选择的分发）。
 * 有可能形成消息死信的循环。为 实例，当队列死信时，可能会发生这种情况 发送到默认交换的消息，而不指定 死信路由密钥。此类循环中的消息（即 两次到达同一队列的邮件）将是 如果整个周期中没有拒绝，则丢弃。
 */
@Slf4j
public class DeadLetterExchanges {

    /**
     * 对消息的死信影响
     * 消息的死信会修改其标头：
     *
     * 交易所名称将替换为最新的死信交易所名称，
     * 路由密钥可以替换为执行死信的队列中指定的密钥，
     * 如果发生上述情况，CC 标头也将被删除，并且
     * 密件抄送标头将根据发件人选择的分发删除
     *
     * 死信进程将数组添加到 每条死信消息都命名为X-Death。 此数组包含每个死信事件的条目， 由一对 {队列， 原因} 标识。 每个这样的条目都是一个包含 的几个字段：
     *
     *      queue：消息在变为死信之前所在的队列的名称
     *      reason：死字的原因，见下文
     *      time：消息作为 64 位 AMQP 0-9-1 时间戳的死信日期和时间
     *      exchange  - 消息发布到的 Exchange （请注意，这将是一个死信 如果消息多次为死信，则交换）
     *      routing：发布消息时使用的路由密钥（包括 CC 密钥，但不包括密件抄送密钥）
     *      count：由于此原因，此消息在此队列中被死信的次数
     *      original-expiration（如果消息由于每条消息的 TTL 而成为死信）： 消息的原始过期属性。过期属性将从 关于死信的消息，以防止它在路由到的任何队列中再次过期。
     * 新条目将附加到 x-death 数组的开头。如果 x-death 已经包含带有 相同的队列和死字原因，其计数字段将是 递增，它将被移动到数组的开头。
     *
     * 原因是描述为什么 消息是死信的，并且是以下之一：
     *
     *      rejected：邮件被拒绝，重新排队参数设置为 false
     *      expired：消息 TTL 已过期
     *      maxlen：已超出允许的最大队列长度
     *      delivery_limit：消息返回的次数超过了限制（由仲裁队列的策略参数传递限制设置）。
     *
     * 为第一个死信添加了三个顶级标头 事件。他们是
     *      x-first-death-reason    第一次的成为死信时的原因，死亡原因分为四种下同分别是：rejected（消息处理被拒绝）、expired （ttl时间到期）、maxlen（超过队列允许最大长度）和Delivery-limit（消息返回的次数超限额）
     *      x-first-death-queue     第一次的成为死信时的队列名称
     *      x-first-death-exchange  第一次的成为死信时的交换器名称
     *
     * 例子:
     *      #contentHeader<basic>(content-type=null, content-encoding=null,
     *      headers={
     *          x-first-death-exchange=exchange_nack,
     *          x-death=[{reason=rejected, count=1, exchange=exchange_nack, time=Thu May 11 18:25:27 CST 2023, routing-keys=[routing_key_nack], queue=queue_nack}],
     *          x-first-death-reason=rejected,
     *          x-first-death-queue=queue_nack},
     *          delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
     */
    @Test
    @SneakyThrows
    public void configurationUsingOptionalQueueArguments() {

        String EXCHANGE_NAME = "exchange_nack";
        String ROUTING_KEY_NAME = "routing_key_nack";
        String QUEUE_NAME = "queue_nack";

        String DEAD_EXCHANGE_NAME = "dead_direct_exchange";
        String DEAD_ROUTING_KEY_NAME = "dead_direct_routing_key";
        String DEAD_QUEUE_NAME = "dead_direct_queue";

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            // 死信交换
            channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(DEAD_QUEUE_NAME, false, false, false, null);
            channel.queueBind(DEAD_QUEUE_NAME, DEAD_EXCHANGE_NAME, DEAD_ROUTING_KEY_NAME);

            // 1.使用者使用 basic.reject 或 basic.nack 对消息进行否定确认，并将重新排队参数设置为 false。
            // 创造死信信息
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(QUEUE_NAME, false, false, false,
                    new HashMap<String, Object>() {{
                        put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
                        put("x-dead-letter-routing-key", DEAD_ROUTING_KEY_NAME);
                    }});
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_NAME);

            for (int i = 0; i < 3; i++) {
                channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_NAME, null, (i + "----").getBytes(StandardCharsets.UTF_8));
                log.info("执行命令 message : {} ", (i + "----"));
            }

            channel.basicConsume(QUEUE_NAME, false,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" ======== [nack] Received '{}'", message);
                        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(1);
            log.info("================================");

            channel.basicConsume(DEAD_QUEUE_NAME, true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" consumerTag : {} ", consumerTag);
                        log.info(" envelope : {} ", delivery.getEnvelope());
                        log.info(" properties : {} ", delivery.getProperties());
                        log.info(" [x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            /**
             * 本轮测试可知 :
             * 已知
             *    1.使用者使用 basic.reject 或 basic.nack 对消息进行否定确认，并将重新排队参数设置为 false。
             *    2.消息由于每条消息的 TTL 而过期;或
             *    3.邮件被丢弃，因为其队列超出了长度限制
             * 2 和 3 已在前面的例子中体现过了，本例将 1 进行了测试，结果均正确
             */
            TimeUnit.SECONDS.sleep(10);

        }

    }

}
